package com.jasonchoi.common.logback;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.Layout;
import ch.qos.logback.core.UnsynchronizedAppenderBase;
import com.cwbase.logback.AdditionalField;
import com.cwbase.logback.JSONEventLayout;
import lombok.Setter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import redis.clients.jedis.JedisPool;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Properties;

/**
 * @Author: JasonChoi
 * @Date: 2020/4/21 15:19
 */
@Setter
public class KafkaAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {
    JedisPool pool;
    JSONEventLayout jsonlayout = new JSONEventLayout();
    Layout<ILoggingEvent> layout;

    private String topic;
    private String bootstrapServers;
    private String valueSerializer;
    private boolean failOnStartup;
    private Producer<String, String> producer;
    private List<String> customProps = new ArrayList<>();

    public KafkaAppender() {
    }

    @Override
    public void start() {
        Objects.requireNonNull(topic, "topic must not be null");
        Objects.requireNonNull(bootstrapServers, "bootstrapServers must not be null");

        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        parseProperties(props);
        try {
            this.producer = new KafkaProducer<>(props);
            super.start();
        } catch (Exception e) {
            if (failOnStartup) {
                addError("Unable to start Kafka Producer", e);
            } else {
                addWarn("Unable to start Kafka Producer", e);
            }
        }
    }

    @Override
    public void stop() {
        super.stop();
        if (Objects.nonNull(this.producer)) {
            this.producer.close();
        }
    }


    @Override
    protected void append(ILoggingEvent event) {
        if (producer != null) {
            try {
                String json = this.layout == null ? this.jsonlayout.doLayout(event) : this.layout.doLayout(event);
                producer.send(new ProducerRecord<>(topic, String.valueOf(Math.random()), json)).get();
            } catch (Exception e) {
                addWarn("Unable to send message to Kafka", e);
            }
        }
    }

    void parseProperties(Properties properties) {
        if (customProps != null) {
            customProps.forEach(property -> {
                String[] p = property.split("\\|");
                if (p.length == 2) {
                    properties.put(p[0], p[1]);
                } else {
                    if (failOnStartup) {
                        addError("Unable to parse property string: " + property);
                    } else {
                        addWarn("Unable to parse property string: " + property);
                    }
                }
            });
        }
    }

    /** @deprecated */
    @Deprecated
    public String getSource() {
        return this.jsonlayout.getSource();
    }

    /** @deprecated */
    @Deprecated
    public void setSource(String source) {
        this.jsonlayout.setSource(source);
    }

    /** @deprecated */
    @Deprecated
    public String getSourceHost() {
        return this.jsonlayout.getSourceHost();
    }

    /** @deprecated */
    @Deprecated
    public void setSourceHost(String sourceHost) {
        this.jsonlayout.setSourceHost(sourceHost);
    }

    /** @deprecated */
    @Deprecated
    public String getSourcePath() {
        return this.jsonlayout.getSourcePath();
    }

    /** @deprecated */
    @Deprecated
    public void setSourcePath(String sourcePath) {
        this.jsonlayout.setSourcePath(sourcePath);
    }

    /** @deprecated */
    @Deprecated
    public String getTags() {
        if (this.jsonlayout.getTags() != null) {
            Iterator<String> i = this.jsonlayout.getTags().iterator();
            StringBuilder sb = new StringBuilder();

            while(i.hasNext()) {
                sb.append((String)i.next());
                if (i.hasNext()) {
                    sb.append(',');
                }
            }

            return sb.toString();
        } else {
            return null;
        }
    }

    /** @deprecated */
    @Deprecated
    public void setTags(String tags) {
        if (tags != null) {
            String[] atags = tags.split(",");
            this.jsonlayout.setTags(Arrays.asList(atags));
        }

    }

    /** @deprecated */
    @Deprecated
    public String getType() {
        return this.jsonlayout.getType();
    }

    /** @deprecated */
    @Deprecated
    public void setType(String type) {
        this.jsonlayout.setType(type);
    }

    /** @deprecated */
    @Deprecated
    public void setMdc(boolean flag) {
        this.jsonlayout.setProperties(flag);
    }

    /** @deprecated */
    @Deprecated
    public boolean getMdc() {
        return this.jsonlayout.getProperties();
    }

    /** @deprecated */
    @Deprecated
    public void setLocation(boolean flag) {
        this.jsonlayout.setLocationInfo(flag);
    }

    /** @deprecated */
    @Deprecated
    public boolean getLocation() {
        return this.jsonlayout.getLocationInfo();
    }

    /** @deprecated */
    @Deprecated
    public void setCallerStackIndex(int index) {
        this.jsonlayout.setCallerStackIdx(index);
    }

    /** @deprecated */
    @Deprecated
    public int getCallerStackIndex() {
        return this.jsonlayout.getCallerStackIdx();
    }

    /** @deprecated */
    @Deprecated
    public void addAdditionalField(AdditionalField p) {
        this.jsonlayout.addAdditionalField(p);
    }

}

